package org.niit.sql

import org.apache.spark.sql.expressions.{Aggregator, MutableAggregationBuffer, UserDefinedAggregateFunction}
import org.apache.spark.sql.types.{DataType, LongType, StructField, StructType}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{DataFrame, Encoder, Encoders, Row, SparkSession, functions}

object SparkSQL_3 {

  /*
    UDF: 用户自定义函数：可以将查询出来的结果进行转换 或者 拼接 等一系列的操作
   */
  def main(args: Array[String]): Unit = {
    //准备环境
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkSQL")
    val spark: SparkSession = SparkSession.builder().config(sparkConf).getOrCreate()
    val sparkContext: SparkContext = spark.sparkContext
    sparkContext.setLogLevel("ERROR")
    //DataFrame读取Json文件
    val df: DataFrame = spark.read.json("input/user.json")
    df.createOrReplaceTempView("user")

    //1.定义UDF函数   第一个参数是定义 自定义UDF函数的名称  第二个参数是 要进行转换或拼接的数据的形参
    spark.udf.register("preName",(name:String)=>{
        "Name:" + name //返回        ↑        ↑
        //name.substring(0,2)        ↑        ↑
    })  //                           ↑        ↑
    //                               ↑        ↑
    //2.使用UDF                     函数名
    spark.sql( "  select preName(username) as username from  user" ).show()
    /*
    +--------+
    |username|
    +--------+
    |Name: zhangsan|
    |Name:lisi|
    |Name:wuwang|
    +--------+
     */


    /*
     2.UDAF  弱类型聚合函数
     */
    //spark.udf.register("ageAvg",new MyAVgUDAF)
   // spark.sql( "select ageAvg(age) from user" ).show()

    /*
     3. UDAF 强类型聚合函数
     */

    spark.udf.register("avgAge",functions.udaf(new MyAvgUDAF))
    spark.sql( "select avgAge(age) from user" ).show()
    spark.close()
  }

  //强类型
  //定义缓冲区       年龄总数            年龄次数
  case class Buff(var total:Long,var count:Long)   // 20 30 40 放到缓冲区  1  1  1 放到缓冲区   临时存储的作用
  class MyAvgUDAF extends Aggregator[Long,Buff,Long]{
    //缓冲区值的初始化
    override def zero: Buff = {
      Buff(0L,0L)
    }
    //根据输入的数据更新缓冲区的数据
    override def reduce(buff: Buff, in: Long): Buff = {
      buff.total = buff.total + in // 0 + 20                // buff :  buff1+20  buff2+30
      buff.count = buff.count + 1
      buff
    }
    //合并缓存区
    override def merge(b1: Buff, b2: Buff): Buff = {

      b1.total = b1.total + b2.total  //  年龄的总和 b1.total = 0 + 20  b2.total = 0 + 30 +40
      b1.count = b1.count + b2.count // 年龄出现的次数 b1.count = 0 + 1  b2.count = 0 + 1 + 1
      b1 // 拥有 年龄总数  年龄次数
    }
    //编写最后的计算功能 根据相应需求编写对应计算功能
    override def finish(buff: Buff): Long = {
      buff.total / buff.count
    }

    override def bufferEncoder: Encoder[Buff] = Encoders.product

    override def outputEncoder: Encoder[Long] = Encoders.scalaLong
  }

//弱类型
 /* class MyAVgUDAF extends UserDefinedAggregateFunction{
       //输入的数据结构 长整型
    override def inputSchema: StructType = {
      StructType(
        Array(
          StructField("age",LongType)
        )
      )
    }
    //缓冲区数据的结构
    override def bufferSchema: StructType = {
      StructType(
        Array(
          StructField("age",LongType),
          StructField("count",LongType),
        )
      )
    }
    //函数计算结果的数据类型
    override def dataType: DataType = LongType

    //函数的稳定性
    override def deterministic: Boolean = true
    //缓冲区的初始化
    override def initialize(buffer: MutableAggregationBuffer): Unit = {
      buffer.update(0,0L)//对 age进行初始化的默认值 0L
      buffer.update(1,0L)//对 count进行初始化的默认值 0L
    }
    //根据输入的值更新缓冲区的数据
    override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
      buffer.update(0,buffer.getLong(0) + input.getLong(0));// 0 +20 +30+10 对年龄的累加
      buffer.update(1,buffer.getLong(1) + 1) // 对次数的累加    0  +1 + 1 + 1 对次数的累加
    }
    //缓冲区进行合并
    override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {
                                                                      // buffer1 : 0:20  30  1:1 + 1
                                                                      // buffer2 : 0:40    1: 1
      buffer1.update(0,   buffer1.getLong(0) + buffer2.getLong(0) )
      buffer1.update(1,   buffer1.getLong(1) + buffer2.getLong(1) )

    }
    //最后进行 相应需求 计算
    override def evaluate(buffer: Row): Any = {
      buffer.getLong(0) / buffer.getLong(1)
    }
  }*/

}
